package com.zx.learn.flink.official.frauddetection.source;

import com.zx.learn.flink.official.frauddetection.entity.Transaction;
import org.apache.flink.streaming.api.functions.source.FromIteratorFunction;

import java.io.Serializable;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

public class TransactionSource extends FromIteratorFunction<Transaction> {
    private static final long serialVersionUID = 1L;
    private static List<Transaction> data = Arrays.asList(new Transaction(1L, 0L, 188.23D), new Transaction(2L, 0L,
            374.79D), new Transaction(3L, 0L, 112.15D), new Transaction(4L, 0L, 478.75D), new Transaction(5L, 0L,
            208.85D), new Transaction(1L, 0L, 379.64D), new Transaction(2L, 0L, 351.44D), new Transaction(3L, 0L,
            320.75D), new Transaction(4L, 0L, 259.42D), new Transaction(5L, 0L, 273.44D), new Transaction(1L, 0L,
            267.25D), new Transaction(2L, 0L, 397.15D), new Transaction(3L, 0L, 0.219D), new Transaction(4L, 0L,
            231.94D), new Transaction(5L, 0L, 384.73D), new Transaction(1L, 0L, 419.62D), new Transaction(2L, 0L,
            412.91D), new Transaction(3L, 0L, 0.77D), new Transaction(4L, 0L, 22.1D), new Transaction(5L, 0L,
            377.54D), new Transaction(1L, 0L, 375.44D), new Transaction(2L, 0L, 230.18D), new Transaction(3L, 0L,
            0.8D), new Transaction(4L, 0L, 350.89D), new Transaction(5L, 0L, 127.55D), new Transaction(1L, 0L,
            483.91D), new Transaction(2L, 0L, 228.22D), new Transaction(3L, 0L, 871.15D), new Transaction(4L, 0L,
            64.19D), new Transaction(5L, 0L, 79.43D), new Transaction(1L, 0L, 56.12D), new Transaction(2L, 0L,
            256.48D), new Transaction(3L, 0L, 148.16D), new Transaction(4L, 0L, 199.95D), new Transaction(5L, 0L,
            252.37D), new Transaction(1L, 0L, 274.73D), new Transaction(2L, 0L, 473.54D), new Transaction(3L, 0L,
            119.92D), new Transaction(4L, 0L, 323.59D), new Transaction(5L, 0L, 353.16D), new Transaction(1L, 0L,
            211.9D), new Transaction(2L, 0L, 280.93D), new Transaction(3L, 0L, 347.89D), new Transaction(4L, 0L,
            459.86D), new Transaction(5L, 0L, 82.31D), new Transaction(1L, 0L, 373.26D), new Transaction(2L, 0L,
            479.83D), new Transaction(3L, 0L, 454.25D), new Transaction(4L, 0L, 83.64D), new Transaction(5L, 0L,
            292.44D));

    public TransactionSource() {
        super(new RateLimitedIterator());
    }

    private static class RateLimitedIterator implements Iterator<Transaction>, Serializable {
        private static final long serialVersionUID = 1L;

        private RateLimitedIterator() {
        }

        @Override
        public boolean hasNext() {
            return Boolean.TRUE;
        }

        @Override
        public Transaction next() {
            try {
                Thread.sleep(100L);
            } catch (InterruptedException var2) {
                throw new RuntimeException(var2);
            }
            int index = (int) (Math.random() * data.size());
            Transaction transaction = data.get(index);
            transaction.setTimestamp(System.currentTimeMillis());
            transaction.setAmount(Math.random() * 1000);
            return transaction;
        }
    }
}
